package com.zyh.kafka.day02;

import com.zyh.kafka.util.KafkaConsumerUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class KafkaConsumerTest2 {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = KafkaConsumerUtils.get("g2");

        consumer.subscribe(Arrays.asList("topic-a","topic-b"));
        boolean boo = true;
        try {
            while(boo){
                ConsumerRecords<String, String> records = consumer.poll(1000);
                if(!records.isEmpty()) {
                    System.out.println("----------遍历所有------------------");
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.println(record);
                    }
                    System.out.println("------------以topic为维度遍历数据--------");
                    Set<String> topics = new HashSet<>();
                    for (TopicPartition partition : records.partitions()) {
                        String topic = partition.topic();
                        topics.add(topic);
                    }
                    for (String topic : topics) {
                        System.out.println("-----------"+topic+"-------------");
                        Iterable<ConsumerRecord<String, String>> iterable = records.records(topic);
                        for (ConsumerRecord<String, String> record : iterable) {
                            System.out.println("record = " + record);
                        }
                    }

                    System.out.println("----------以分区为维度--------");
                    Set<TopicPartition> partitions = records.partitions();
                    for (TopicPartition partition : partitions) {
                        System.out.println("------------"+partition+"------");
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("record = " + record);
                        }
                    }
                }
            }
        } catch (Exception e) {
            boo = false;
            e.printStackTrace();
        } finally {
            KafkaConsumerUtils.close(consumer);
        }
    }
}
